#include "znx_event_loop.h"
#include "comm/znx_handles.h"


#define MAX_POLLER_WAIT_TIMEOUT_MS              (64)
#define TIMER_LAZY_DELAY_MS                     (5)
#define EVENT_LOOP_NEVENTS                      (8192)


volatile znx_bool_t                             proc_exit = ZNX_FALSE;


static void
znx_event_loop_notify(znx_event_loop_data_t *loop_data)
{
    uint64_t counter = 1;
    znx_syscall_write(loop_data->notify_conn.fd, &counter, sizeof(counter));
}


static void
znx_event_loop_notify_handler(znx_event_t *event)
{
    uint64_t        counter;
    znx_conn_t      *conn = event->data;

    znx_syscall_result_t res;
    for ( ;; ) {
        res = znx_syscall_read(conn->fd, &counter, sizeof(counter));
        if (res.eno != 0) {
            break;  // EAGAIN or other error, ignore.
        }
    }

    event->ready = 0;
}


static void
znx_event_loop_init(znx_event_loop_data_t *loop_data)
{
    // epoll_create:
    // since Linux 2.6.8, the size argment is ignored, but must be greater than zero.
    // see man epoll_create
    znx_syscall_result_t res = znx_syscall_epoll_create(EVENT_LOOP_NEVENTS);
    if (res.eno != 0) {
        ZNX_DEF_LOG_FATAL("epoll_create failed, errno: %d", (int)res.eno);
    }

    znx_cleanup_list_add(&loop_data->cleanup_list, znx_fd_close_handle, (void *)(intptr_t)res.res);
    loop_data->epfd = (int)res.res;

    loop_data->event_list = znx_thread_calloc(sizeof(znx_epoll_event_t) * EVENT_LOOP_NEVENTS);
    znx_cleanup_list_add(&loop_data->cleanup_list, znx_thread_free_handle, loop_data->event_list);

    res = znx_syscall_eventfd(0, EFD_NONBLOCK|EFD_CLOEXEC);
    if (res.eno != 0) {
        ZNX_DEF_LOG_FATAL("eventfd failed, errno: %d", (int)res.eno);
    }

    loop_data->notify_conn.fd = (int)res.res;
    loop_data->notify_conn.read.handler = znx_event_loop_notify_handler;
    znx_cleanup_list_add(&loop_data->cleanup_list,
        znx_event_loop_remove_conn_handle, &loop_data->notify_conn);

    if (!znx_event_loop_start_io(&loop_data->notify_conn)) {
        ZNX_DEF_LOG_FATAL("start notify conn io failed");
    }

    if (!znx_event_loop_enable_read(&loop_data->notify_conn)) {
        ZNX_DEF_LOG_FATAL("enabled notify conn read io failed");
    }

    if (loop_data->init_handle) {
        loop_data->init_handle(loop_data);
    }
}


static void
znx_event_loop_move_task_to_local(znx_event_loop_data_t *loop_data)
{
    znx_spinlock_lock(&loop_data->lock);
    znx_queue_move(&loop_data->external_queue, &loop_data->local_queue);
    znx_spinlock_unlock(&loop_data->lock);
}


static void
znx_event_loop_proc_local_queue(znx_event_loop_data_t *loop_data)
{
    znx_queue_t             *node;
    znx_dispatch_event_t    *dispatch_event;

    for ( ;; ) {
        if (znx_queue_empty(&loop_data->local_queue)) {
            return;
        }

        node = znx_queue_last(&loop_data->local_queue);
        dispatch_event = znx_queue_data(node, znx_dispatch_event_t, queue);
        znx_queue_remove(node);
        dispatch_event->handler(dispatch_event);
    }
}


static void
znx_event_loop_exec_expire_timers(znx_event_loop_data_t *loop_data)
{
    znx_event_t         *event;
    znx_rbtree_node_t   *node;

    for ( ;; ) {
        if (znx_rbtree_empty(&loop_data->timer_rbtree)) {
            return;
        }

        node = znx_rbtree_min(loop_data->timer_rbtree.root, loop_data->timer_rbtree.sentinel);
        event = znx_rbtree_data(node, znx_event_t, timer);
        if (event->time_key - znx_usec_2_msec(get_cache_time()) > 0) {
            return;
        }

        znx_rbtree_delete(&loop_data->timer_rbtree, node);

        event->timer_set = 0;
        event->timedout = 1;
        event->handler(event);
    }
}


// @return: how long does the earliest timer expire, utils: msecond
// If return 0, mean has timer expired, -1 mean not has timer task.
static int64_t
znx_event_loop_earliest_expire_ms(znx_event_loop_data_t *loop_data)
{
    int64_t                 ms;
    znx_rbtree_node_t       *node;
    znx_event_t             *event;

    if (znx_rbtree_empty(&loop_data->timer_rbtree)) {
        return -1;
    }

    node = znx_rbtree_min(loop_data->timer_rbtree.root,
        loop_data->timer_rbtree.sentinel);
    event = znx_rbtree_data(node, znx_event_t, timer);
    ms = event->time_key - znx_usec_2_msec(get_cache_time());

    return ms > 0 ? ms : 0;
}


static int64_t
znx_event_loop_calculation_wait_timeout(znx_event_loop_data_t *loop_data)
{
    znx_bool_t has_posted_task;
    znx_spinlock_lock(&loop_data->lock);
    has_posted_task = !znx_queue_empty(&loop_data->external_queue);
    znx_spinlock_unlock(&loop_data->lock);

    if (has_posted_task) {
        return 0;
    }

    if (!znx_queue_empty(&loop_data->local_queue)) {
        return 0;
    }

    int64_t wait_timeout_ms =
        znx_event_loop_earliest_expire_ms(loop_data);
    if (wait_timeout_ms == 0) {
        return 0;
    }

    if (!znx_queue_empty(&loop_data->ready_queue)) {
        return 0;
    }

    if (wait_timeout_ms < 0 || wait_timeout_ms > MAX_POLLER_WAIT_TIMEOUT_MS) {
        return MAX_POLLER_WAIT_TIMEOUT_MS;
    }

    return wait_timeout_ms;
}


// proc ready queue task
static void
znx_event_loop_proc_ready_events(znx_event_loop_data_t *loop_data)
{
    znx_queue_t     *node;
    znx_event_t     *event;

    for ( ;; ) {
        if (znx_queue_empty(&loop_data->ready_queue)) {
            return;
        }

        node = znx_queue_last(&loop_data->ready_queue);
        event = znx_queue_data(node, znx_event_t, queue);
        znx_queue_remove(node);

        if (event->ready) {
            event->handler(event);
        }
    }
}


static void
znx_event_loop_exec_pollor_events(znx_event_loop_data_t *loop_data,
    int64_t wait_timeout_msec)
{
    int             events;
    uint32_t        revents;
    znx_event_t      *rev, *wev;
    znx_conn_t       *c;

    znx_syscall_result_t res = znx_syscall_epoll_wait(
        loop_data->epfd, loop_data->event_list,
        EVENT_LOOP_NEVENTS, (int)wait_timeout_msec);
    if (res.eno != 0 || res.res == 0) {     // EINTR or ETIMEDOUT
        return;
    }

    events = (int)res.res;
    for (int i = 0; i < events; i++) {
        c = (znx_conn_t *)loop_data->event_list[i].data.ptr;
        if (c->fd == INVALID_FD) {
            ZNX_DEF_LOG_FATAL("epoll: stale conn %p, invalid fd", c);
            continue;
        }

        revents = loop_data->event_list[i].events;
        if (revents & (EPOLLERR|EPOLLHUP)) {
            ZNX_DEF_LOG_WARN("epoll_wait() error on fd: %d ev: %d", c->fd, revents);
            // If the error events were returned,
            // add EPOLLIN and EPOLLOUT to handle the events.
            revents |= EPOLLIN|EPOLLOUT;
        }

        rev = &c->read;
        if (revents & EPOLLIN) {
            rev->ready = 1;
            // ready & active, but event not in queue.
            if (rev->active && rev->queue.prev == NULL) {
                znx_queue_insert_head(&loop_data->ready_queue, &rev->queue);
            }
        }

        wev = &c->write;
        if (revents & EPOLLOUT) {
            wev->ready = 1;
            // ready & active, but event not in queue.
            if (wev->active && wev->queue.prev == NULL) {
                znx_queue_insert_head(&loop_data->ready_queue, &wev->queue);
            }
        }
    }

    znx_event_loop_proc_ready_events(loop_data);
}


static void
znx_event_loop_executer(znx_event_loop_data_t *loop_data)
{
    znx_event_loop_move_task_to_local(loop_data);
    znx_event_loop_proc_local_queue(loop_data);

    get_latest_time();
    znx_event_loop_exec_expire_timers(loop_data);

    znx_event_loop_proc_ready_events(loop_data);

    int64_t wait_timeout_msec =
        znx_event_loop_calculation_wait_timeout(loop_data);
    znx_event_loop_exec_pollor_events(loop_data, wait_timeout_msec);
}


static znx_bool_t
znx_event_loop_can_stop(znx_event_loop_data_t *loop_data)
{
    if (loop_data->async_tasks != 0) {
        return ZNX_FALSE;
    }

    if (!znx_queue_empty(&loop_data->ready_queue)) {
        return ZNX_FALSE;
    }

    if (!znx_rbtree_empty(&loop_data->timer_rbtree)) {
        return ZNX_FALSE;
    }

    if (!znx_queue_empty(&loop_data->local_queue)) {
        return ZNX_FALSE;
    }

    int res;
    znx_spinlock_lock(&loop_data->lock);
    res = znx_queue_empty(&loop_data->external_queue);
    znx_spinlock_unlock(&loop_data->lock);

    if (!res) {
        return ZNX_FALSE;
    }

    return ZNX_TRUE;
}


void *
znx_event_loop_runner(znx_thread_base_t *thread_base)
{
    znx_event_loop_data_t *loop_data = thread_base->data;
    znx_event_loop_init(loop_data);

    for ( ;; ) {
        if (proc_exit) {
            if (loop_data->proc_exit_handle) {
                loop_data->proc_exit_handle(loop_data);
            }

            if (znx_event_loop_can_stop(loop_data)) {
                znx_atomic_bool_set_true(&loop_data->can_stop);
            }
        }

        if (znx_atomic_bool_is_true(&loop_data->quit)) {
            if (!loop_data->stop) {
                if (loop_data->quit_handle) {
                    loop_data->quit_handle(loop_data);
                }
                loop_data->stop = 1;
            }
        }

        if (loop_data->stop && znx_atomic_bool_is_true(&loop_data->can_stop)) {
            break;
        }

        znx_event_loop_executer(loop_data);
    }

    znx_cleanup_list_exec(&loop_data->cleanup_list);

    znx_conn_close(&loop_data->notify_conn);

    znx_atomic_bool_set_true(&loop_data->stopped);

    return NULL;
}


void
znx_event_loop_quit(znx_thread_base_t *thread_base)
{
    znx_event_loop_data_t *loop_data = thread_base->data;
    znx_atomic_bool_set_true(&loop_data->quit);
}


void
znx_event_loop_post_event(znx_thread_base_t *thread_base,
    znx_dispatch_event_t *dispatch_event)
{
    znx_event_loop_data_t *loop_data = thread_base->data;
    znx_bool_t external_queue_empty;

    // if post thread == current thread, we can post event to local queue.
    if (thread_base == current_thread_base) {
        znx_queue_insert_head(&loop_data->local_queue, &dispatch_event->queue);
        return;
    }

    znx_spinlock_lock(&loop_data->lock);
    external_queue_empty = znx_queue_empty(&loop_data->external_queue);
    znx_queue_insert_head(&loop_data->external_queue, &dispatch_event->queue);
    znx_spinlock_unlock(&loop_data->lock);

    if (!external_queue_empty) {
        return;
    }

    znx_event_loop_notify(loop_data);
}


void
znx_event_loop_add_timer(znx_event_t *event, int64_t timeout_ms)
{
    znx_event_loop_data_t *loop_data = (znx_event_loop_data_t *)current_thread_base->data;
    int64_t time_key = znx_usec_2_msec(get_cache_time()) + timeout_ms;

    if (event->timer_set) {
        if (ZNX_ABS(time_key - event->time_key) < TIMER_LAZY_DELAY_MS) {
            return;
        }
        znx_event_loop_del_timer(event);
    }

    event->time_key = time_key;
    znx_rbtree_insert(&loop_data->timer_rbtree, &event->timer);
    event->timer_set = 1;
}


void
znx_event_loop_del_timer(znx_event_t *event)
{
    if (!event->timer_set) {
        return;
    }

    znx_event_loop_data_t *loop_data = current_thread_base->data;
    znx_rbtree_delete(&loop_data->timer_rbtree, &event->timer);
    event->timer_set = 0;
}


znx_bool_t
znx_event_loop_start_io(znx_conn_t *conn)
{
    if (conn->start) {
        return ZNX_TRUE;
    }
    conn->start = 1;

    znx_event_loop_data_t *loop_data = current_thread_base->data;

    if (conn->flag != ZNX_CONN_EPOLL_ET) {
        return ZNX_TRUE;
    }

    conn->ee.events |= EPOLLIN;
    conn->ee.events |= EPOLLOUT;
    conn->ee.events |= EPOLLET;

    znx_syscall_result_t res =
        znx_syscall_epoll_ctl(loop_data->epfd, EPOLL_CTL_ADD, conn->fd, &conn->ee);
    if (res.eno != 0) {
        ZNX_DEF_LOG_ERROR("add epoll events failed, epfd: %d, fd: %d, flag: %d, event: %d, errno: %d",
            loop_data->epfd, conn->fd, conn->flag, conn->ee.events, res.eno);
        return ZNX_FALSE;
    }

    return ZNX_TRUE;
}


znx_bool_t
znx_event_loop_stop_io(znx_conn_t *conn)
{
    if (!conn->start) {
        return ZNX_TRUE;
    }
    conn->start = 0;

    znx_event_loop_data_t *loop_data = current_thread_base->data;

    uint32_t old_events = conn->ee.events;
    conn->ee.events &= ~(uint32_t)EPOLLIN;
    conn->ee.events &= ~(uint32_t)EPOLLOUT;
    conn->ee.events &= ~(uint32_t)EPOLLET;
    if (old_events == conn->ee.events) {
        return ZNX_TRUE;
    }

    znx_syscall_result_t res =
        znx_syscall_epoll_ctl(loop_data->epfd, EPOLL_CTL_DEL, conn->fd, &conn->ee);
    if (res.eno != 0) {
        ZNX_DEF_LOG_ERROR("remove epoll evenets failed, fd: %d, flag: %d, eno: %d",
            conn->fd, conn->flag, res.eno);
        return ZNX_FALSE;
    }

    conn->read.active = 0;
    conn->write.active = 0;

    if (conn->read.queue.prev) {
        znx_queue_remove(&conn->read.queue);
    }

    if (conn->write.queue.prev) {
        znx_queue_remove(&conn->write.queue);
    }

    return ZNX_TRUE;
}


znx_bool_t
znx_event_loop_enable_io(znx_conn_t *conn, znx_bool_t write)
{
    if (!conn->start) {
        ZNX_DEF_LOG_FATAL("must start_io first");
    }

    znx_event_loop_data_t *loop_data = current_thread_base->data;

    uint32_t old_event = conn->ee.events;
    if (conn->flag != ZNX_CONN_EPOLL_ET) {
        uint32_t event = EPOLLIN;
        if (write) {
            event = EPOLLOUT;
        }
        conn->ee.events |= event;
    }

    if (old_event != conn->ee.events) {
        znx_syscall_result_t res =
            znx_syscall_epoll_ctl(loop_data->epfd,
                old_event == 0 ? EPOLL_CTL_ADD : EPOLL_CTL_MOD, conn->fd, &conn->ee);
        if (res.eno != 0) {
            ZNX_DEF_LOG_ERROR("epoll_ctl: add or mod event failed, fd: %d, flag: %d, eno: %d",
                conn->fd, conn->flag, res.eno);
            return ZNX_FALSE;
        }
    }

    znx_event_t *ev = &conn->read;
    if (write) {
        ev = &conn->write;
    }

    ev->active = 1;

    // event ready, but not in ready queue.
    if (ev->ready == 1 && ev->queue.prev == NULL) {
        znx_queue_insert_head(&loop_data->ready_queue, &ev->queue);
    }

    // event no ready, but in ready queue.
    if (ev->ready == 0 && ev->queue.prev != NULL) {
        znx_queue_remove(&ev->queue);
    }

    return ZNX_TRUE;
}


static znx_bool_t
znx_event_loop_disable_io(znx_conn_t *conn, znx_bool_t write)
{
    if (!conn->start) {
        ZNX_DEF_LOG_FATAL("must start_io first");
    }

    znx_event_loop_data_t *loop_data = current_thread_base->data;

    uint32_t old_event = conn->ee.events;
    if (conn->flag != ZNX_CONN_EPOLL_ET) {
        uint32_t event = EPOLLIN;
        if (write) {
            event = EPOLLOUT;
        }
        conn->ee.events &= ~event;
    }

    if (old_event != conn->ee.events) {
        znx_syscall_result_t res =
            znx_syscall_epoll_ctl(loop_data->epfd,
                conn->ee.events == 0 ? EPOLL_CTL_DEL : EPOLL_CTL_MOD, conn->fd, &conn->ee);
        if (res.eno != 0) {
            ZNX_DEF_LOG_ERROR("epoll_ctl: del or mod event failed, fd: %d, flag: %d, eno: %d",
                conn->fd, conn->flag, res.eno);
            return ZNX_FALSE;
        }
    }

    znx_event_t *ev = &conn->read;
    if (write) {
        ev = &conn->write;
    }

    ev->active = 0;
    if (ev->queue.prev != NULL) {
        znx_queue_remove(&ev->queue);
    }

    return ZNX_TRUE;
}


znx_bool_t
znx_event_loop_enable_read(znx_conn_t *conn)
{
    return znx_event_loop_enable_io(conn, ZNX_FALSE);
}


znx_bool_t
znx_event_loop_disable_read(znx_conn_t *conn)
{
    return znx_event_loop_disable_io(conn, ZNX_FALSE);
}


znx_bool_t
znx_event_loop_enable_write(znx_conn_t *conn)
{
    return znx_event_loop_enable_io(conn, ZNX_TRUE);
}


znx_bool_t
znx_event_loop_disable_write(znx_conn_t *conn)
{
    return znx_event_loop_disable_io(conn, ZNX_TRUE);
}


void
znx_event_loop_remove_conn(znx_conn_t *conn)
{
    znx_event_loop_stop_io(conn);

    // remove read events in conn
    if (conn->read.timer_set) {
        znx_event_loop_del_timer(&conn->read);
    }

    // remove write events in conn
    if (conn->write.timer_set) {
        znx_event_loop_del_timer(&conn->write);
    }
}

